package com.zimmor.mq.rocket_train.transzction;

import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.HashMap;
import java.util.Map;

/**
 * @Author:zimmor
 * @Date: 2023/2/16 9:38
 * @Description 发送事务消息
 * @version: 1.0.0
 */

public class Producer {
    public static void main(String[] args) throws MQClientException {
        //创建发送事务消息的生产者
        TransactionMQProducer transactionMQProducer = new TransactionMQProducer("producer-group7");
        //设置namesrv
        transactionMQProducer.setNamesrvAddr("192.168.204.136:9876;192.168.204.137:9876");
        //设置事务消息监听器
        transactionMQProducer.setTransactionListener(new TransactionListener() {
            Map result = new HashMap();
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                LocalTransactionState state = LocalTransactionState.UNKNOW;
                result = (Map) JSON.parse(message.getBody());
                result.put("transactionID", message.getTransactionId());
                //执行某个事务，根据结果判断返回的状态
                try{
                    boolean status = false;
//                    int i = 1/0;
                    if(status) {
                        state = LocalTransactionState.COMMIT_MESSAGE;
                    }
                } catch (Exception e) {
                    state = LocalTransactionState.ROLLBACK_MESSAGE;
                }
                return state;
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                Object transactionID = result.get("transactionID");
                LocalTransactionState state = LocalTransactionState.UNKNOW;
                try {
                    int i = 1/0;
                    if (transactionID != null) {
                        state = LocalTransactionState.COMMIT_MESSAGE;
                    } else {
                        state = LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                }catch(Exception e) {
                    e.printStackTrace();
                }
                return state;
            }
        });
        //启动生产者
        transactionMQProducer.start();
        //发送消息
        Map order = new HashMap();
        order.put("orderNo","1234556");
        order.put("username", "张三");
        Message message = new Message("transaction", "tag3", JSON.toJSONString(order).getBytes());
        TransactionSendResult transactionSendResult = transactionMQProducer.sendMessageInTransaction(message, null);
        System.out.println("msg:" + transactionSendResult.toString());
    }
}
